package org.dyz.banana.service;

import org.dyz.banana.controllers.TestProducerController;
import org.dyz.banana.vo.JsonResponse;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

@Service
public class KafkaProducerServiceImpl {

    private static final Logger log= LoggerFactory.getLogger(KafkaProducerServiceImpl.class);

    @Autowired
    private KafkaTemplate<String,Object> kafkaTemplate;


    public JsonResponse sendMessage(String topic,String message){
        if(topic==null){
            String errMsg = "topic不能为空！";
            log.error(errMsg);
            return JsonResponse.fail(errMsg);
        }else if(message==null){
            String errMsg = "message不能为空！";
            log.error(errMsg);
            return JsonResponse.fail(errMsg);
        }else{
            kafkaTemplate.send(topic, message).addCallback(success->{
                String retTopic = success.getProducerRecord().topic();
                int partition = success.getRecordMetadata().partition();
                long offset = success.getRecordMetadata().offset();

                log.info("message:"+message+" 发送成功！topic="+topic+"; partition="+partition+"; offset="+offset);

            },failure->{
                log.info("发送失败！原因："+failure.getMessage());
            });
            return JsonResponse.success();
        }
    }
}
